package com.shujia.spark.kafka

import java.util.Properties

import kafka.consumer.{Consumer, ConsumerConfig, KafkaStream}
import kafka.serializer.{Decoder, StringDecoder}
import kafka.utils.VerifiableProperties

object Demo2Consumer {
  def main(args: Array[String]): Unit = {

    val config: Properties = new Properties()

    //zookeeper 配置，通过zk 可以负载均衡的获取broker
    config.put("zookeeper.connect", "node1:2181")

    //group 代表一个消费组,  如果每次想要读取所有的数据，每次都需要修改消费者组
    config.put("group.id", "asdasd")

    //zk连接超时时间
    config.put("zookeeper.session.timeout.ms", "10000")

    //zk同步时间
    config.put("zookeeper.sync.time.ms", "200")

    //自动提交间隔时间
    config.put("auto.commit.interval.ms", "1000")

    //消息日志自动偏移量,防止宕机后数据无法读取


    //smallest
    //当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
    //largest
    //当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
    //none
    //topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

    config.put("auto.offset.reset", "smallest")

    //序列化类
    config.put("serializer.class", "kafka.serializer.StringEncoder")


    //构建consumer connection 对象
    val consumer = Consumer.create(new ConsumerConfig(config))



    //指定需要消费的topic  指定消费线程数
    val topics = Map("topic" -> 3)


    //指定key的编码格式
    val keyDecoder: Decoder[String] = new StringDecoder(new VerifiableProperties)
    //指定value的编码格式
    val valueDecoder: Decoder[String] = new StringDecoder(new VerifiableProperties)

    //获取topic 和 接受到的stream 集合
    val maps: collection.Map[String, List[KafkaStream[String, String]]] = consumer.createMessageStreams(topics, keyDecoder, valueDecoder)

    //取出一个topic所有的流对象
    val list: List[KafkaStream[String, String]] = maps("topic")

    for (stream <- list) {

      //每一个流启动一个线程消费
      new Thread(new Runnable {
        override def run(): Unit = {

          //读取stream  中的数据
          for (lineMsg <- stream.iterator()) {
            val value = lineMsg.message()
            val key = lineMsg.key()
            val partitions = lineMsg.partition
            val offset = lineMsg.offset

            println(s"$key\t$value\t$partitions\t$offset")

          }
        }
      }).start()//启动线程
    }



  }
}
